package process

import bean.SensorReading
import org.apache.flink.api.common.state._
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.{KeyedProcessFunction, ProcessFunction}
import org.apache.flink.streaming.api.scala.{DataStream, OutputTag, StreamExecutionEnvironment}
import org.apache.flink.util.Collector

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/27 12:14
  */
object SideOutputTest {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //file source
    val inputPath: String = "src/main/resources/sensor.txt"
    val fileStream: DataStream[String] = env.readTextFile(inputPath)
    val socketStream = env.socketTextStream("192.168.88.106", 7777)

    //先转换为特定的类型
    val dataStream: DataStream[SensorReading] = socketStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

    val highTempStream: DataStream[SensorReading] = dataStream
      .process(new SplitTempProcessor(30.0))

    highTempStream.print("high")
    highTempStream.getSideOutput(new OutputTag[(String, Long, Double)]("low")).print("low")


    env.execute("sideoutput  test")
  }

  //实现自定义的KeyProcessFunction
  class SplitTempProcessor(threshold: Double) extends ProcessFunction[SensorReading, SensorReading] {
    override def processElement(i: SensorReading, context: ProcessFunction[SensorReading, SensorReading]#Context, collector: Collector[SensorReading]): Unit = {
      if (i.temperature > threshold){
        //如果当前温度值大于30，那么输出到主流
        collector.collect(i)
      }else{
        //如果不超过30度，那么输出到侧输出流
        context.output(new OutputTag[(String, Long, Double)]("low"),(i.id,i.timestamp,i.temperature))
      }
    }
  }

}
